package com.leilei.join.interval;

import com.leilei.join.common.*;
import com.leilei.join.window.VehicleDetailWatermark;
import com.leilei.join.window.VehicleInfoWatermark;
import com.leilei.join.window.VehicleTypeWatermark;
import com.leilei.join.window.VehicleWatermark;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/27 21:41
 * @desc 不使用window 进行Join
 */
public class FlinkIntervalJoin {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        // 获取车辆数据
        SingleOutputStreamOperator<Vehicle> vehicleDs = env.addSource(new VehicleSource())
                .assignTimestampsAndWatermarks(new VehicleWatermark());

        // 获取车辆明细数据
        SingleOutputStreamOperator<VehicleDetail> vehicleDetailDs = env.addSource(new VehicleDetailSource())
                .assignTimestampsAndWatermarks(new VehicleDetailWatermark());

        // 车辆数据与 车辆明细数据关联
        SingleOutputStreamOperator<VehicleInfo> vehicleInfoDs = vehicleDs.keyBy(Vehicle::getId)
                //关联条件  vehicle中Id=VehicleDetail 中 vehicleId
                .intervalJoin(vehicleDetailDs.keyBy(VehicleDetail::getVehicleId))
                // 可看作过滤条件 车辆明细时间时间-5秒 <= 车辆时间时间 <= 车辆明细时间时间+5秒
                .between(Time.seconds(-5), Time.seconds(5))
                .process(new VehicleJoinDetailProcessFunction());

        // 获取车辆类型数据
        SingleOutputStreamOperator<Tuple2<Integer, String>> vehicleTypeDs = env.addSource(new VehicleTypeSource())
                .assignTimestampsAndWatermarks(new VehicleTypeWatermark());

        //再将 vehicleInfo 与车辆类型数据关联
        SingleOutputStreamOperator<Tuple4<Integer, String, Integer, String>> result = vehicleInfoDs.keyBy(VehicleInfo::getType)
                .intervalJoin(vehicleTypeDs.keyBy(t -> t.f0))
                .between(Time.seconds(-5), Time.seconds(5))
                .process(new VehicleJoinTypeProcessFunction());

        result.print("最终车辆数据:");
        env.execute("interval-join");
    }
}
